package com.vhall.component.service.vote.impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper;
import com.baomidou.mybatisplus.core.toolkit.CollectionUtils;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.vhall.component.dao.vote.RoomVoteLkMapper;
import com.vhall.component.dao.vote.VoteAnswersMapper;
import com.vhall.component.dao.vote.VoteOptionCountMapper;
import com.vhall.component.dao.vote.VotesMapper;
import com.vhall.component.entity.room.dto.RoomJoinsEntityDTO;
import com.vhall.component.entity.room.dto.RoomsEntityDTO;
import com.vhall.component.entity.vote.RoomVoteLkEntity;
import com.vhall.component.entity.vote.VoteAnswersEntity;
import com.vhall.component.entity.vote.VoteOptionCountEntity;
import com.vhall.component.entity.vote.VotesEntity;
import com.vhall.component.entity.vote.vo.VoteDetailReqVO;
import com.vhall.component.entity.vote.vo.VoteDetailRspVO;
import com.vhall.component.entity.vote.vo.VotesListRspVO;
import com.vhall.component.entity.vote.vo.room.*;
import com.vhall.component.framework.common.exception.BusinessException;
import com.vhall.component.framework.common.exception.GlobalErrorCode;
import com.vhall.component.plugin.common.constant.PaasCommonConstant;
import com.vhall.component.plugin.common.constant.RedisKey;
import com.vhall.component.plugin.common.constant.RoomConstant;
import com.vhall.component.plugin.common.exception.BizErrorCode;
import com.vhall.component.plugin.common.support.VssToken;
import com.vhall.component.plugin.common.support.VssTokenThreadLocal;
import com.vhall.component.plugin.core.service.OssService;
import com.vhall.component.plugin.core.service.PaasService;
import com.vhall.component.service.notice.NoticesService;
import com.vhall.component.service.room.RoomInnerService;
import com.vhall.component.service.vote.*;
import com.vhall.component.service.vote.delay.DelayEvent;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.ObjectUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.ReturnType;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.io.ByteArrayInputStream;
import java.io.InputStream;
import java.nio.charset.StandardCharsets;
import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.TimeUnit;

/**
 * @author longzhen.gong
 * @date 2021/7/7 11:09
 */
@Service
@Slf4j
public class VoteRoomServiceImpl implements VoteRoomService {

    @Autowired
    private RoomVoteLkMapper roomVoteLkMapper;

    @Autowired
    private RoomInnerService roomInnerService;

    @Autowired
    private VotesMapper votesMapper;

    @Autowired
    private VoteService voteService;

    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    @Autowired
    private PaasService paasService;

    @Autowired
    private VotesRepository votesRepository;

    @Autowired
    private VoteAnswersMapper voteAnswersMapper;


    @Autowired
    private NoticesService noticesService;

    @Autowired
    private OssService ossService;

    @Autowired
    private VoteOptionCountMapper voteOptionCountMapper;

    @Autowired
    private ObjectMapper objectMapper;

    @Autowired
    private VoteCacheService voteCacheService;

    private static final String VOTE_ID_KEY = "vote_id";

    private static final String ACCOUNT_ID_KEY = "account_id";

    @Autowired
    private DelayQueue<DelayEvent> voteDelayQueue;
    @Override
    public Boolean delete(Long voteId, Integer accountId) {

        RoomVoteLkEntity roomVoteLkEntities = roomVoteLkMapper.selectOne(Wrappers.<RoomVoteLkEntity>lambdaQuery()
                .eq(RoomVoteLkEntity::getVoteId, voteId));
        if (roomVoteLkEntities == null) {
            throw new BusinessException(BizErrorCode.COMP_VOTE_INVALID);
        }
        checkRoomRoleName(accountId, roomVoteLkEntities.getRoomId());

        roomVoteLkMapper.deleteById(roomVoteLkEntities.getId());
        getVote(voteId, roomVoteLkEntities.getRoomId());
        delRunningVoteIdCache(roomVoteLkEntities.getRoomId(), voteId);
        votesMapper.deleteById(voteId);

        Integer count = roomVoteLkMapper.selectCount(Wrappers.<RoomVoteLkEntity>lambdaQuery()
                .eq(RoomVoteLkEntity::getRoomId, roomVoteLkEntities.getRoomId()));
        RoomsEntityDTO roomInfo = roomInnerService.getRoomInfo(roomVoteLkEntities.getRoomId());
        if (count == 0){
            Map<String, Object> body = new HashMap<>(8);
            body.put("type", "vote_empty");
            body.put("room_id", roomInfo.getRoomId());
            paasService.sendMessage(roomInfo.getChannelId(), JSON.toJSONString(body));
        }else {
            Long vote = lastVote(roomVoteLkEntities.getRoomId());
            if (vote != null){
                Map<String, Object> body = new HashMap<>(8);
                body.put("type", "vote_last_finished");
                body.put("room_id", roomInfo.getRoomId());
                body.put("vote_id", vote);
                paasService.sendMessage(roomInfo.getChannelId(), JSON.toJSONString(body));
            }else {
                Map<String, Object> body = new HashMap<>(8);
                body.put("type", "vote_empty");
                body.put("room_id", roomInfo.getRoomId());
                paasService.sendMessage(roomInfo.getChannelId(), JSON.toJSONString(body));
            }
        }
        return true;
    }
    private Long lastVote(String roomId) {
        LambdaQueryWrapper<RoomVoteLkEntity> queryWrapper = Wrappers.lambdaQuery();
        queryWrapper.eq(RoomVoteLkEntity::getRoomId, roomId);
        queryWrapper.eq(RoomVoteLkEntity::getPublish, 1);
        queryWrapper.orderByDesc(RoomVoteLkEntity::getUpdatedAt);
        queryWrapper.last(" limit 1");
        RoomVoteLkEntity roomVoteLkEntity = roomVoteLkMapper.selectOne(queryWrapper);
        if (roomVoteLkEntity != null) {
            return roomVoteLkEntity.getVoteId();
        }
        return null;
    }
    @Override
    public Boolean unBindRoom(Long voteId, String roomId) {
        VotesEntity votesEntity = votesMapper.selectById(voteId);
        if (votesEntity == null) {
            throw new BusinessException(BizErrorCode.COMP_VOTE_INVALID);
        }
        RoomsEntityDTO rooms = roomInnerService.getRoomInfo(roomId);
        if (rooms == null) {
            throw new BusinessException(BizErrorCode.EMPTY_ROOM);
        }
        if (ObjectUtils.notEqual(rooms.getAccountId(), votesEntity.getAccountId())) {
            throw new BusinessException(BizErrorCode.COMP_VOTE_INVALID);
        }
        RoomVoteLkEntity roomVoteLkEntity = roomVoteLkMapper.selectOne(Wrappers.<RoomVoteLkEntity>lambdaQuery()
                .eq(RoomVoteLkEntity::getRoomId, rooms.getRoomId())
                .eq(RoomVoteLkEntity::getVoteId, votesEntity.getVoteId()));
        if (roomVoteLkEntity == null) {
            throw new BusinessException(BizErrorCode.COMP_VOTE_INVALID);
        }
        roomVoteLkEntity.setBind(VoteConstants.RoomVoteLkBind.UNBIND);
        return roomVoteLkMapper.updateById(roomVoteLkEntity) > 0;
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public CopyRspVO copy(CopyReqVO reqVO, Integer accountId) {
        checkRoomRoleName(accountId, reqVO.getRoomId());

        RoomsEntityDTO rooms = roomInnerService.getRoomInfo(reqVO.getRoomId());
        if (rooms == null) {
            throw new BusinessException(BizErrorCode.EMPTY_ROOM);
        }
        VotesEntity votesEntity = votesMapper.selectById(reqVO.getVoteId());
        if (votesEntity == null) {
            throw new BusinessException(BizErrorCode.COMP_VOTE_INVALID);
        }
        JSONObject formInfo = paasService.getFormInfo(reqVO.getVoteId() + "");
        JSONArray formDetail = formInfo.getJSONArray("detail");
        if (CollectionUtils.isEmpty(formDetail)) {
            throw new BusinessException(BizErrorCode.COMP_VOTE_INVALID);
        }
        formDetail.forEach(voteDetail -> {
            JSONObject qu = (JSONObject) voteDetail;
            qu.remove("id");
            qu.remove("third_party_user_id");
            qu.remove("created_at");
            qu.remove("app_id");
        });
        Map<String, Object> param = new HashMap<>(16);
        param.put("title", formInfo.getString("title"));
        param.put("description", formInfo.getString("description"));
        param.put("publish", "Y");
        param.put("detail", formDetail.toJSONString());
        param.put("owner_id", accountId);
        JSONObject newVote = paasService.createForm(param);
        if (newVote.get("id") == null) {
            throw new BusinessException(BizErrorCode.COMP_VOTE_INVALID);
        }

        votesEntity.setVoteId(newVote.getLong("id"));
        votesEntity.setTitle(votesEntity.getTitle());
        votesEntity.setExtend(votesEntity.getExtend());
        votesEntity.setAccountId(votesEntity.getAccountId());
        votesEntity.setIsPublic(votesEntity.getIsPublic());
        votesEntity.setLimitTime(votesEntity.getLimitTime());
        votesEntity.setOptionNum(votesEntity.getOptionNum());
        votesEntity.setCreatedAt(LocalDateTime.now());
        votesEntity.setUpdatedAt(LocalDateTime.now());
        votesRepository.save(votesEntity);
        voteService.bindRoom(votesEntity.getVoteId(), rooms.getRoomId());
        CopyRspVO copyRspVO = new CopyRspVO();
        BeanUtils.copyProperties(votesEntity, copyRspVO);
        return copyRspVO;
    }


    /**
     * 删除正在进行中的投票 ID 缓存
     *
     * @param roomId
     * @param voteId
     */
    @Override
    public void delRunningVoteIdCache(String roomId, Long voteId) {
        String lua = " local runVoteId = redis.call('hget', KEYS[1], ARGV[1])\n" +
                "        if runVoteId == ARGV[2] then\n" +
                "            redis.call('hdel', KEYS[1], ARGV[1])\n" +
                "        end\n" +
                "        return runVoteId";
        stringRedisTemplate.execute((RedisCallback) connection -> {
            connection.eval(lua.getBytes(StandardCharsets.UTF_8), ReturnType.VALUE, 1,
                    (RedisKey.INTERACT_TOOL + roomId).getBytes(StandardCharsets.UTF_8),
                    RedisKey.INTERACT_TOOL_FILED.getBytes(StandardCharsets.UTF_8),
                    String.valueOf(voteId).getBytes(StandardCharsets.UTF_8));
            return null;
        });
    }

    /**
     * 提交答案
     *
     * @param reqVO
     * @return
     */
    @Override
    public Boolean answer(AnswerReqVO reqVO) {
        try {
            JSON.parseArray(reqVO.getExtend());
        } catch (Exception e) {
            throw new BusinessException(BizErrorCode.COMP_VOTE_EXTEND);
        }
        Long voteId = reqVO.getVoteId();
        String lockKey = "vote_answer_" + reqVO.getRoomId() + voteId + reqVO.getAccountId();
        if (Boolean.FALSE.equals(stringRedisTemplate.opsForValue().setIfAbsent(lockKey, "1", 86400, TimeUnit.SECONDS))) {
            throw new BusinessException(BizErrorCode.COMP_VOTE_SUBMIT_REPEAT);
        }
        VotesEntity votesEntity = voteCacheService.getVote(voteId);
        if (votesEntity == null) {
            throw new BusinessException(BizErrorCode.COMP_VOTE_INVALID);
        }
        RoomVoteLkEntity roomVoteLkEntity = voteCacheService.getRoomVoteLk(reqVO.getRoomId(), voteId);
        if (roomVoteLkEntity == null) {
            throw new BusinessException(BizErrorCode.COMP_VOTE_INVALID);

        }
        if (roomVoteLkEntity.getPublish().equals(VoteConstants.RoomVoteLkBind.PUBLISH_NO)) {
            throw new BusinessException(BizErrorCode.COMP_VOTE_INVALID);
        }
        if (roomVoteLkEntity.getIsFinish() == VoteConstants.RoomVoteLkBind.FINISH) {
            throw new BusinessException(BizErrorCode.COMP_VOTE_FINISHED);
        }

        ObjectNode objectNode;
        String newJson;
        try {
            objectNode = objectMapper.readValue(objectMapper.writeValueAsString(reqVO), ObjectNode.class);
            objectNode.put("vote_lk_id", roomVoteLkEntity.getId());
            objectNode.put("option_num", votesEntity.getOptionNum());
            objectNode.put(ACCOUNT_ID_KEY, reqVO.getAccountId());
            newJson = objectMapper.writeValueAsString(objectNode);
        } catch (JsonProcessingException e) {
            log.error("json转换错误", e);
            throw new BusinessException(GlobalErrorCode.FAILED);
        }

        // redis list leftpop异步存储在服务器宕机或者重启等情况下容易出现数据丢失现象
        /*stringRedisTemplate.opsForZSet().add(RedisKey.QUEUE_VOTE_ANSWER_ROOMS, reqVO.getRoomId(), LocalDateTime.now().toEpochSecond(ZoneOffset.of("+8")));
        stringRedisTemplate.opsForList().rightPush(String.format(RedisKey.QUEUE_VOTE_ANSWER_LIST_KEY, reqVO.getRoomId()), newJson);*/
        // 同步存储
        answerToDb(newJson);

        // stringRedisTemplate.expire(lockKey, 0, TimeUnit.SECONDS);
        return true;
    }

    @Override
    public void answerToDb(String param) {
        JSONObject answerJson = JSON.parseObject(param);
        Long answerId = answerJson.getLong("answer_id");
        VoteAnswersEntity voteAnswersEntity = voteCacheService.getVoteAnswer(answerId);
        if (voteAnswersEntity != null) {
            log.info("投票-queue:{}", param);
            throw new BusinessException(BizErrorCode.COMP_VOTE_ANSWERED);
        }
        voteAnswersEntity = new VoteAnswersEntity();
        voteAnswersEntity.setVoteId(answerJson.getLong(VOTE_ID_KEY));
        voteAnswersEntity.setAnswerId(answerId);
        voteAnswersEntity.setRoomId(answerJson.getString("room_id"));
        voteAnswersEntity.setExtend(answerJson.getString("extend"));
        voteAnswersEntity.setAccountId(answerJson.getInteger(ACCOUNT_ID_KEY));

        RoomJoinsEntityDTO roomJoinsEntityDTO = roomInnerService
                .getRoomJoins(answerJson.getInteger(ACCOUNT_ID_KEY), answerJson.getString("room_id"));
        if (roomJoinsEntityDTO == null) {
            log.info("用户未找到-vote-queue:{}", param);
            throw new BusinessException(BizErrorCode.EMPTY_VOTE);
        }
        String extend = answerJson.getString("extend");
        Long voteLkId = answerJson.getLong("vote_lk_id");
        Integer optionNum = answerJson.getInteger("option_num");
        dealExtend(voteLkId, extend, optionNum);
        voteAnswersEntity.setJoinId(roomJoinsEntityDTO.getJoinId());
        voteAnswersMapper.insert(voteAnswersEntity);
        roomInnerService.updateIsAnsweredVoteByJoinId(voteAnswersEntity.getRoomId(), voteAnswersEntity.getAccountId() + "");
    }

    @Override
    public Boolean publish(PublishReqVO reqVO) {
        VssToken vssToken = VssTokenThreadLocal.getToken();
        Integer thirdPartyUserId = vssToken.getThirdPartyUserId();
        Object vote = stringRedisTemplate.opsForHash().get(RedisKey.INTERACT_TOOL + reqVO.getRoomId(),
                RedisKey.INTERACT_TOOL_FILED);
        if (vote != null) {
            throw new BusinessException(BizErrorCode.COMP_VOTE_RUNNING_NOT_PUBLISH);
        }
        VotesEntity votesEntity = votesMapper.selectById(reqVO.getVoteId());
        if (votesEntity == null) {
            throw new BusinessException(BizErrorCode.COMP_VOTE_INVALID);
        }
        RoomsEntityDTO rooms = roomInnerService.getRoomInfo(reqVO.getRoomId());
        if (rooms == null) {
            throw new BusinessException(BizErrorCode.EMPTY_ROOM);
        }

        RoomJoinsEntityDTO roomJoinsEntityDTO = roomInnerService.getRoomJoins(thirdPartyUserId, reqVO.getRoomId());
        if (roomJoinsEntityDTO == null
                || roomJoinsEntityDTO.getRoleName().equals(RoomConstant.AUDIENCE)) {
            throw new BusinessException(BizErrorCode.BIZ_NO_PERMISSION);
        }
        RoomVoteLkEntity roomVoteLkEntity = roomVoteLkMapper.selectOne(Wrappers.<RoomVoteLkEntity>lambdaQuery()
                .eq(RoomVoteLkEntity::getRoomId, reqVO.getRoomId())
                .eq(RoomVoteLkEntity::getVoteId, reqVO.getVoteId()));

        // 发布投票，投票信息本地静态化
        JSONObject extend = new JSONObject();
        extend.put("limit_time", votesEntity.getLimitTime());
        String ossFileUrl = writeInfoLocal(votesEntity.getVoteId(), "vote", extend);

        LocalDateTime finishTime = LocalDateTime.ofEpochSecond(
                System.currentTimeMillis() / 1000 + votesEntity.getLimitTime(),
                0, ZoneOffset.ofHours(8));

        if (roomVoteLkEntity == null) {
            roomVoteLkEntity = new RoomVoteLkEntity();
            roomVoteLkEntity.setRoomId(rooms.getRoomId());
            roomVoteLkEntity.setBind(VoteConstants.RoomVoteLkBind.BIND);
            roomVoteLkEntity.setVoteId(votesEntity.getVoteId());
            roomVoteLkEntity.setAccountId(rooms.getAccountId());
            roomVoteLkEntity.setExtend(votesEntity.getExtend());
            roomVoteLkEntity.setPublish(VoteConstants.RoomVoteLkBind.PUBLISH_YES);
            //是否发布评分 0-未发布 1-发布
            roomVoteLkEntity.setIsRelease(VoteConstants.RoomVoteLkBind.UNRELEASE);
            roomVoteLkEntity.setIsFinish(VoteConstants.RoomVoteLkBind.UNFINISH);
            roomVoteLkEntity.setFinishTime(finishTime);
            roomVoteLkMapper.insert(roomVoteLkEntity);
        } else if (roomVoteLkEntity.getPublish() == VoteConstants.RoomVoteLkBind.PUBLISH_YES) {
            throw new BusinessException(BizErrorCode.COMP_VOTE_INVALID);
        } else {
            roomVoteLkEntity.setPublish(VoteConstants.RoomVoteLkBind.PUBLISH_YES);
            roomVoteLkEntity.setFinishTime(finishTime);
            roomVoteLkEntity.setAccountId(rooms.getAccountId());
            roomVoteLkMapper.updateById(roomVoteLkEntity);
        }
        Map<String, Object> body = new HashMap<>(8);
        body.put("type", "vote_push");
        body.put(PaasCommonConstant.VOTE_ID_KEY, votesEntity.getVoteId());
        body.put(PaasCommonConstant.NICK_NAME_KEY, roomJoinsEntityDTO.getNickname());
        body.put(PaasCommonConstant.ROOM_JOIN_ID_KEY, roomJoinsEntityDTO.getAccountId());
        body.put(PaasCommonConstant.ROOM_ROLE_KEY, roomJoinsEntityDTO.getRoleName());
        body.put("info_url", ossFileUrl);
        body.put("start_time", LocalDateTime.now().toEpochSecond(ZoneOffset.of("+8")));
        body.put("limit_time", votesEntity.getLimitTime());

        paasService.sendMessage(rooms.getChannelId(), JSON.toJSONString(body));

        stringRedisTemplate.opsForHash().put(RedisKey.INTERACT_TOOL + reqVO.getRoomId(), RedisKey.INTERACT_TOOL_FILED, votesEntity.getVoteId() + "");

        //发送公告
//        noticesService.sendNotice(reqVO.getRoomId(), votesEntity.getVoteId() + "", roomJoinsEntityDTO.getAccountId(), "vote_push");
        if (votesEntity.getLimitTime() > 0){
            log.info("添加投票到延时队列,房间:{},投票:{}", rooms.getIlId(), votesEntity.getVoteId());
            DelayEvent event = new DelayEvent(votesEntity.getVoteId(), votesEntity.getLimitTime(), rooms.getRoomId());
            event.setAccountId(thirdPartyUserId);
            voteDelayQueue.add(event);
            stringRedisTemplate.opsForList().rightPush(RedisKey.VOTE_DELAY_QUEUE, JSON.toJSONString(event));
        }
        // 投递任务，定时结束队列
        stringRedisTemplate.opsForZSet().add(RedisKey.VOTE_FINISH_RLKID_QUEUE, roomVoteLkEntity.getId() + "", roomVoteLkEntity.getFinishTime().toEpochSecond(ZoneOffset.of("+8")));

        return true;
    }

    /**
     * 取消投票发布
     *
     * @param reqVO
     * @return
     */
    @Override
    public Boolean cancelPublish(PublishReqVO reqVO) {

        Long voteId = Long.valueOf(reqVO.getVoteId());
        String roomId = reqVO.getRoomId();
        VssToken vssToken = VssTokenThreadLocal.getToken();
        Integer thirdPartyUserId = vssToken.getThirdPartyUserId();

        checkRoomRoleName(thirdPartyUserId, roomId);

        getVote(voteId, roomId);


        RoomVoteLkEntity roomVoteLkEntity = roomVoteLkMapper.selectOne(Wrappers.<RoomVoteLkEntity>lambdaQuery()
                .eq(RoomVoteLkEntity::getRoomId, reqVO.getRoomId())
                .eq(RoomVoteLkEntity::getVoteId, reqVO.getVoteId()));
        if (roomVoteLkEntity == null) {
            throw new BusinessException(BizErrorCode.COMP_VOTE_INVALID);
        }

        roomVoteLkEntity.setPublish(VoteConstants.RoomVoteLkBind.PUBLISH_NO);
        roomVoteLkEntity.setFinishTime(null);
        roomVoteLkMapper.updateById(roomVoteLkEntity);

        // 删除缓存的投票 ID
        delRunningVoteIdCache(roomId, voteId);

        return true;
    }

    @Override
    public VotesListRspVO getLastVote(String roomId) {
        return null;
    }

    private String writeInfoLocal(Long voteId, String module, JSONObject extend) {
        String ossFileUrl;
        /*
         * 投票只会有一次发布，并且发布之后不允许修改投票信息，在发布时进行oss文件保存并保存到缓存中，后面再来请求，取缓存
         */
        String key = RedisKey.VOTE_INFO_OSS_URL + voteId;
        String value = stringRedisTemplate.opsForValue().get(key);
        if (StringUtils.isNotEmpty(value)) {
            return value;
        }
        JSONObject detail = paasService.getFormInfo(voteId + "");
        detail.putAll(extend);
        String type = "text/plain";
        String name = "upload/" + module + "/" + voteId + ".json";
        byte[] bytes = detail.toString().getBytes(StandardCharsets.UTF_8);
        InputStream stream = new ByteArrayInputStream(bytes);
        ossFileUrl = ossService.upload(stream, name, type);
        stringRedisTemplate.opsForValue().set(key, ossFileUrl, 60 * 60 * 24, TimeUnit.SECONDS);
        return ossFileUrl;
    }

    private boolean dealExtend(Long rvlkId, String extend, Integer optionNum) {

        List<VoteOptionCountEntity> optionList = voteService.getVoteOptionCountInfoByRvlkId(rvlkId);
        if (CollectionUtils.isEmpty(optionList)) {
            throw new BusinessException(BizErrorCode.COMP_VOTE_OPTION_INVALID);
        }
        Map<Object, Object> optionsMap = stringRedisTemplate.opsForHash().entries(RedisKey.VOTE_RVLK + rvlkId);
        JSONArray extendArr = JSON.parseArray(extend);

        String replys_key = "replys";
        for (Object obj : extendArr) {
            JSONObject answer = (JSONObject) obj;
            //投票对应表单问题id
            String questionId = answer.getString("id");
            //验证投票提交问题id与投票问题id一致
//            if (optionsMap.get(questionId) == null) {
//                throw new BusinessException(BizErrorCode.COMP_VOTE_CONTENT_INVALID);
//            }
            if (StringUtils.isBlank(answer.getString(replys_key))) {
                continue;
            }
            switch (answer.getString("type")) {
                case "radio":
                    UpdateWrapper<VoteOptionCountEntity> wrappers = Wrappers.update();
                    wrappers.eq("rvlk_id", rvlkId);
                    stringRedisTemplate.opsForHash().increment(RedisKey.VOTE_QUESTION + questionId, answer.getString(replys_key), 1);
                    wrappers.setSql("count = count + 1");
                    wrappers.eq("`option`", answer.getString(replys_key));
                    voteOptionCountMapper.update(null, wrappers);
                    break;
                case "checkbox":
                    List<String> replysList = answer.getJSONArray(replys_key).toJavaList(String.class);
                    if (replysList.isEmpty()) {
                        log.info("选项不能为空,{}", replysList);
                        return false;
                    }
                    if (replysList.size() > optionNum) {
                        log.error("选项超出规范数量rvlkId:{},extend:{},optionNum:{}", rvlkId, extend, optionNum);
                        return false;
                    }
                    replysList.forEach(r -> stringRedisTemplate.opsForHash().increment(RedisKey.VOTE_QUESTION + questionId, r, 1));
                    voteOptionCountMapper.updateBatchCount(replysList, rvlkId);
                    break;
                default:
                    break;
            }
        }
        return true;

    }


    @Override
    public Boolean voteFinish(FinishReqVO reqVO) {
        Long voteId = reqVO.getVoteId();
        String roomId = reqVO.getRoomId();
        getVote(voteId, roomId);
        VssToken vssToken = VssTokenThreadLocal.getToken();
        Integer thirdPartyUserId = vssToken.getThirdPartyUserId();

        RoomJoinsEntityDTO roomJoinsEntityDTO = checkRoomRoleName(thirdPartyUserId, roomId);

        RoomVoteLkEntity roomVoteLkEntity = roomVoteLkMapper.selectOne(Wrappers.<RoomVoteLkEntity>lambdaQuery()
                .eq(RoomVoteLkEntity::getRoomId, reqVO.getRoomId())
                .eq(RoomVoteLkEntity::getVoteId, reqVO.getVoteId()));
        if (roomVoteLkEntity == null) {
            throw new BusinessException(BizErrorCode.COMP_VOTE_INVALID);
        }
        if (roomVoteLkEntity.getPublish() != VoteConstants.RoomVoteLkBind.PUBLISH_YES) {
            throw new BusinessException(BizErrorCode.COMP_VOTE_NOT_PUBLISH);
        }
        if (roomVoteLkEntity.getIsFinish() == VoteConstants.RoomVoteLkBind.FINISH) {
            throw new BusinessException(BizErrorCode.COMP_VOTE_FINISHED);
        }
        roomVoteLkEntity.setIsFinish(VoteConstants.RoomVoteLkBind.FINISH);
        roomVoteLkEntity.setFinishTime(LocalDateTime.now());
        roomVoteLkEntity.setUpdatedAt(LocalDateTime.now());
        roomVoteLkMapper.updateById(roomVoteLkEntity);

        //发送结束
        Map<String, Object> body = new HashMap<>(16);
        body.put("type", "vote_finish");
        body.put(PaasCommonConstant.VOTE_ID_KEY, voteId);
        body.put(PaasCommonConstant.NICK_NAME_KEY, roomJoinsEntityDTO.getNickname());
        body.put(PaasCommonConstant.ROOM_JOIN_ID_KEY, roomJoinsEntityDTO.getAccountId());
        body.put(PaasCommonConstant.ROOM_ROLE_KEY, roomJoinsEntityDTO.getRoleName());
        RoomsEntityDTO roomInfo = roomInnerService.getRoomInfo(roomId);

        paasService.sendMessage(roomInfo.getChannelId(), JSON.toJSONString(body));

        //发送公告
//        noticesService.sendNotice(roomId, voteId + "", roomJoinsEntityDTO.getAccountId(), "vote_finish");

        // 删除缓存的投票 ID
        delRunningVoteIdCache(roomId, voteId);

        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        PushReqVO pushReqVO = new PushReqVO();
        pushReqVO.setRoomId(reqVO.getRoomId());
        pushReqVO.setVoteId(reqVO.getVoteId());
        votePush(pushReqVO);

        return true;
    }

    @Override
    public Boolean votePush(PushReqVO reqVO) {
        Long voteId = reqVO.getVoteId();
        String roomId = reqVO.getRoomId();
        //判断是否消费完 answer是否都落库了
        if (stringRedisTemplate.opsForList().size(RedisKey.VOTE_ANSWER_LIST_QUEUE + reqVO.getRoomId()) > 0) {
            throw new BusinessException(BizErrorCode.COMP_VOTE_BUSINESS_RESULT_STATING);
        }

        VssToken vssToken = VssTokenThreadLocal.getToken();
        Integer thirdPartyUserId = vssToken.getThirdPartyUserId();

        RoomJoinsEntityDTO roomJoinsEntityDTO = checkRoomRoleName(thirdPartyUserId, roomId);

        getVote(voteId, roomId);

        RoomVoteLkEntity roomVoteLkEntity = roomVoteLkMapper.selectOne(Wrappers.<RoomVoteLkEntity>lambdaQuery()
                .eq(RoomVoteLkEntity::getRoomId, reqVO.getRoomId())
                .eq(RoomVoteLkEntity::getVoteId, reqVO.getVoteId()));

        roomVoteLkEntity.setIsRelease(VoteConstants.RoomVoteLkBind.RELEASE);
        roomVoteLkMapper.updateById(roomVoteLkEntity);
        VoteDetailReqVO voteDetailReqVO = new VoteDetailReqVO();
        voteDetailReqVO.setVoteId(voteId);
        voteDetailReqVO.setRoomId(roomId);
        voteDetailReqVO.setAccountId(thirdPartyUserId);
        voteDetailReqVO.setForm("1");
        List<VoteDetailRspVO> voteDetailRspVO = voteService.voteDetail(voteDetailReqVO, thirdPartyUserId);

        Map<String, Object> body = new HashMap<>(8);
        body.put("type", "vote_final_statis");
        body.put(PaasCommonConstant.VOTE_ID_KEY, voteId);
        body.put(PaasCommonConstant.NICK_NAME_KEY, roomJoinsEntityDTO.getNickname());
        body.put(PaasCommonConstant.ROOM_JOIN_ID_KEY, roomJoinsEntityDTO.getAccountId());
        body.put(PaasCommonConstant.ROOM_ROLE_KEY, roomJoinsEntityDTO.getRoleName());
        body.put("vote_static", voteDetailRspVO);
        RoomsEntityDTO roomInfo = roomInnerService.getRoomInfo(reqVO.getRoomId());
        paasService.sendMessage(roomInfo.getChannelId(), JSON.toJSONString(body));
        // 设置公布结果标识
        stringRedisTemplate.opsForValue().set(RedisKey.VOTE_PUSH_FLAG + voteId, "1", 1,TimeUnit.DAYS);
//        noticesService.sendNotice(roomId, voteId + "", roomJoinsEntityDTO.getAccountId(), "vote_final_statis");

        return true;
    }

    /**
     * 是否提交过投票
     *
     * @param reqVO
     * @return
     */
    @Override
    public CheckSurveyRspVO checkSurvey(CheckSurveyReqVO reqVO) {
        CheckSurveyRspVO checkSurveyRspVO = new CheckSurveyRspVO();

        Long voteId = reqVO.getVoteId();
        String roomId = reqVO.getRoomId();
        VssToken vssToken = VssTokenThreadLocal.getToken();
        Integer thirdPartyUserId = vssToken.getThirdPartyUserId();
        RoomJoinsEntityDTO roomJoinsEntityDTO = roomInnerService.getRoomJoins(thirdPartyUserId, roomId);
        if (roomJoinsEntityDTO == null) {
            throw new BusinessException(BizErrorCode.BIZ_NO_PERMISSION);
        }

        RoomVoteLkEntity roomVoteLk = voteCacheService.getRoomVoteLk(roomId, voteId);
        if (Objects.isNull(roomVoteLk)) {
            throw new BusinessException(BizErrorCode.COMP_VOTE_INVALID);
        }
        VoteAnswersEntity voteAnswersEntity = voteAnswersMapper.selectOne(
                Wrappers.<VoteAnswersEntity>lambdaQuery().eq(VoteAnswersEntity::getVoteId, voteId)
                        .eq(VoteAnswersEntity::getRoomId, roomId)
                        .eq(VoteAnswersEntity::getJoinId, roomJoinsEntityDTO.getJoinId()));

        checkSurveyRspVO.setIsAnswer(voteAnswersEntity != null);
        checkSurveyRspVO.setIsFinish(
                roomVoteLk.getIsFinish() == VoteConstants.RoomVoteLkBind.FINISH || (roomVoteLk.getFinishTime() != null && LocalDateTime
                        .now().isAfter(roomVoteLk.getFinishTime())));

        return checkSurveyRspVO;
    }

    /**
     * 投票实时统计结果推送(不发公告)
     *
     * @param reqVO
     * @return
     */
    @Override
    public Boolean pushStatis(PushStatisReqVO reqVO) {
        VssToken vssToken = VssTokenThreadLocal.getToken();
        Integer thirdPartyUserId = vssToken.getThirdPartyUserId();
        Long voteId = reqVO.getVoteId();
        String roomId = reqVO.getRoomId();

        RoomJoinsEntityDTO roomJoinsEntityDTO = checkRoomRoleName(thirdPartyUserId, roomId);

        getVote(voteId, roomId);

        VoteDetailReqVO voteDetailReqVO = new VoteDetailReqVO();
        BeanUtils.copyProperties(reqVO, voteDetailReqVO);
        voteDetailReqVO.setForm("1");
        voteDetailReqVO.setAccountId(thirdPartyUserId);
        List<VoteDetailRspVO> voteDetailRspVos = voteService.voteDetail(voteDetailReqVO, thirdPartyUserId);
        Map<String, Object> body = new HashMap<>(8);
        body.put("type", "vote_statis");
        body.put(PaasCommonConstant.VOTE_ID_KEY, voteId.toString());
        body.put(PaasCommonConstant.NICK_NAME_KEY, roomJoinsEntityDTO.getNickname());
        body.put(PaasCommonConstant.ROOM_JOIN_ID_KEY, roomJoinsEntityDTO.getAccountId().toString());
        body.put(PaasCommonConstant.ROOM_ROLE_KEY, Integer.valueOf(roomJoinsEntityDTO.getRoleName()));
        body.put("vote_static", voteDetailRspVos);
        RoomsEntityDTO roomInfo = roomInnerService.getRoomInfo(reqVO.getRoomId());
        paasService.sendMessage(roomInfo.getChannelId(), JSON.toJSONString(body));
        return true;
    }

    private VotesEntity getVote(Long voteId, String roomId) {
        VotesEntity votesEntity = votesMapper.selectById(voteId);
        if (votesEntity == null) {
            throw new BusinessException(BizErrorCode.COMP_VOTE_INVALID);
        }
        RoomsEntityDTO rooms = roomInnerService.getRoomInfo(roomId);
        if (rooms == null) {
            throw new BusinessException(BizErrorCode.EMPTY_ROOM);
        }
//        if (ObjectUtils.notEqual(rooms.getAccountId(), votesEntity.getAccountId())) {
//            throw new BusinessException(BizErrorCode.COMP_VOTE_INVALID);
//        }
        return votesEntity;
    }


    private RoomJoinsEntityDTO checkRoomRoleName(Integer accountId, String roomId){
        RoomJoinsEntityDTO roomJoinsEntityDTO = roomInnerService.getRoomJoins(accountId, roomId);
        if (roomJoinsEntityDTO == null
                || roomJoinsEntityDTO.getRoleName().equals(RoomConstant.AUDIENCE)) {
            throw new BusinessException(BizErrorCode.BIZ_NO_PERMISSION);
        }
        return roomJoinsEntityDTO;
    }

}
